package com.guaoran.study.odps;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONPObject;
import com.aliyun.odps.*;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.task.SQLTask;
import lombok.extern.slf4j.Slf4j;

import java.sql.DriverManager;
import java.sql.SQLException;
import java.text.DecimalFormat;
import java.util.List;

/**
 * @author : gu cheng
 * @description : odps demo
 * @date : 2021/5/6 14:19
 */
@Slf4j
public class OppsDemo {
    static String odpsUrl = "http://service.odps.aliyun.com/api";
    static String accessId = "LTAI5t62B8m3jT6dKCwU7kmk";
    static String accessKey = "l5azALY6Xw860v8OF7DJJ3P3OKCPfE";
    static String defaultProject = "default_project_5C3A256";

    static String odpsUrlTemp = odpsUrl;
    static String accessIdTemp = accessId;
    static String accessKeyTemp = accessKey;
    static String defaultProjectTemp = defaultProject;

    /**
     * 获取所有表元数据
     * @param odps
     */
    public static void getTablesInfo(Odps odps){
        for (Table t : odps.tables()) {
            System.out.println("表名:"+t.getName()
                    +"\t注释:"+t.getComment()
                    +"\t查看表所占磁盘的物理大小:"+t.getPhysicalSize()
                    +"\t获取内部存储大小，单位：Byte:"+getNetFileSizeDescription(t.getSize())
            );
        }
    }

    /**
     * 转换字节
     * @param bytes
     * @return
     */
    public static String getNetFileSizeDescription(long bytes) {
        if (bytes == 0) {
            return "0B";
        }
        final String[] units = new String[]{"B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"};
        int digitGroups = (int) (Math.log10(bytes) / Math.log10(1024));
        return new DecimalFormat("#,##0.##").format(bytes / Math.pow(1024, digitGroups)) + "" + units[digitGroups];

    }


    /**
     * 获取表字段元数据
     * @param odps
     */
    public static void getColumnsInfo(Odps odps){
        for (Table t : odps.tables()) {
            List<Column> columns = t.getSchema().getColumns();
            for (Column c : columns) {
                System.out.println("表名:"+t.getName()
                        +"\t表注释:"+t.getComment()
                        +"\t字段名:"+c.getName()
                        +"\t字段注释:"+c.getComment()
                        +"\t字段类型:"+c.getTypeInfo()
                );
            }
        }
    }


    public static Odps getOdps(){
        Account account = new AliyunAccount(accessId, accessKey);
        Odps odps = new Odps(account);
        odps.setEndpoint(odpsUrl);
        odps.setDefaultProject(defaultProject);
        return odps;
    }

    public static Odps getOdpsTemp(){
        Account account = new AliyunAccount(accessIdTemp, accessKeyTemp);
        Odps odps = new Odps(account);
        odps.setEndpoint(odpsUrlTemp);
        odps.setDefaultProject(defaultProjectTemp);
        return odps;
    }


    public static boolean testConnectJDBC(){
        try {
            Class.forName("com.aliyun.odps.jdbc.OdpsDriver");
            DriverManager.getConnection("jdbc:odps:"+odpsUrl+"?project="+defaultProject, accessId, accessKey);
            return true;
        } catch (ClassNotFoundException | SQLException e) {
            e.printStackTrace();
        }
        return false;
    }

    public static boolean testConnect(){
        Odps odps = getOdps();
        System.out.println("1");
        for (Table table:odps.tables()){
            break;
        }
        System.out.println("2");
        return false;
    }
    public static void main(String[] args) throws OdpsException {
        Odps odps = getOdpsTemp();
        List<Column> sale_detail = odps.tables().get("sale_detail").getSchema().getColumns();
        for (Column column : sale_detail) {
            System.out.println(JSONObject.toJSONString(column));
        }
        testQuery(odps);
        // 获取所有表元数据
        // getTablesInfo(odps);

        // 获取表字段元数据
        // getColumnsInfo(odps);

    }

    public static void testQueryForThread(Odps odps, String sql) {
        try {
            Instance run = SQLTask.run(odps, sql);
            run.waitForSuccess();
            List<Record> result = SQLTask.getResult(run);
            Column[] columns = result.get(0).getColumns();
            StringBuilder sb = new StringBuilder();
            for (Column column : columns) {
                sb.append(column.getName()).append("\t");
            }
            log.info("column desc:{}",sb.toString());

            for (Record record : result) {
                sb = new StringBuilder();

                for (int i = 0; i < columns.length; i++) {
                    sb.append(record.get(i)).append("\t");
                }
                log.info("value desc:{}",sb.toString());
            }
        } catch (Exception e){
            log.error("Exception:",e);
        }
    }

    public static void testQuery(Odps odps){
        Long start = System.currentTimeMillis();
        for (Table t : odps.tables()) {
            String sql = "select * from "+ t.getName() +" limit 2 ;";
            new Thread(new QueryThread(odps,sql),"thread_name:"+t.getName()).start();
        }
        log.info("执行结束总耗时：{}毫秒",System.currentTimeMillis()-start);
    }

    static class QueryThread implements Runnable{
        private Odps odps;
        private String sql;

        public QueryThread(Odps odps, String sql) {
            this.odps = odps;
            this.sql = sql;
        }

        @Override
        public void run() {
            OppsDemo.testQueryForThread(odps,sql);
        }
    }
}
